package com.cl.spark.node;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cl.spark.base.BaseSparkNode;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.cl.spark.enums.SparkNodeEnum;
import org.springframework.stereotype.Component;
import scala.annotation.meta.param;

import java.util.*;
// todo
@Component
public class GroupNode extends BaseSparkNode {

    @Override
    public SparkResult process(SparkParam sparkParam) {
        return SparkResult.success(group(sparkParam));
    }

    public static JSONArray group(SparkParam sparkParam) {

        JSONObject expression = sparkParam.getNodeExpression();
        String statisticsMethod = expression.getString("statisticsMethod");

        if ("求和".equals(statisticsMethod)) {
            return sum(sparkParam);
        } else if ("计数".equals(statisticsMethod)) {
            return count(sparkParam);
        } else if ("平均值".equals(statisticsMethod)) {
            return avg(sparkParam);
        }
        return new JSONArray();
    }

    public static JSONArray sum(SparkParam sparkParam) {
        JSONArray res = new JSONArray();
        List<JSONArray> dataList = sparkParam.getDataList();
        JSONArray data = dataList.get(0);
        JSONObject expression = sparkParam.getNodeExpression();
        JSONArray groupFields = expression.getJSONArray("groupFields");
        JSONArray statisticsFields = expression.getJSONArray("statisticsFields");
        String groupType = expression.getString("groupType");
        String splitChar = expression.getString("splitChar");


        Map<String, JSONArray> groupFieldsAndData = collectGroupFieldsAndData(data, groupFields, groupType, splitChar);

        for (Map.Entry<String, JSONArray> entry : groupFieldsAndData.entrySet()) {
            JSONArray groupData = entry.getValue();
            JSONObject resItem = groupData.getJSONObject(0);
            for (int i = 0; i < groupData.size(); i++) {
                JSONObject item = groupData.getJSONObject(i);
                for (int j = 0; j < statisticsFields.size(); j++) {
                    String statisticsField = statisticsFields.getString(j);
                    String newField = statisticsField + "_sum";
                    resItem.put(newField, resItem.getIntValue(newField) + item.getIntValue(statisticsField));
                }
            }
            res.add(resItem);
        }

        return res;
    }

    public static JSONArray count(SparkParam sparkParam) {
        JSONArray res = new JSONArray();
        List<JSONArray> dataList = sparkParam.getDataList();
        JSONArray data = dataList.get(0);
        JSONObject expression = sparkParam.getNodeExpression();
        JSONArray groupFields = expression.getJSONArray("groupFields");
        JSONArray statisticsFields = expression.getJSONArray("statisticsFields");
        String groupType = expression.getString("groupType");
        String splitChar = expression.getString("splitChar");

        Map<String, JSONArray> groupFieldsAndData = collectGroupFieldsAndData(data, groupFields, groupType, splitChar);

        for (Map.Entry<String, JSONArray> entry : groupFieldsAndData.entrySet()) {
            JSONArray groupData = entry.getValue();
            JSONObject resItem = groupData.getJSONObject(0);
            for (int i = 0; i < groupData.size(); i++) {
                for (int j = 0; j < statisticsFields.size(); j++) {
                    String statisticsField = statisticsFields.getString(j);
                    String newField = statisticsField + "_count";
                    resItem.put(newField, resItem.getIntValue(newField) + 1);
                }
            }
            res.add(resItem);
        }

        return res;
    }

    public static JSONArray avg(SparkParam sparkParam) {
        JSONArray res = new JSONArray();
        List<JSONArray> dataList = sparkParam.getDataList();
        JSONArray data = dataList.get(0);
        JSONObject expression = sparkParam.getNodeExpression();
        JSONArray groupFields = expression.getJSONArray("groupFields");
        JSONArray statisticsFields = expression.getJSONArray("statisticsFields");
        String groupType = expression.getString("groupType");
        String splitChar = expression.getString("splitChar");

        Map<String, JSONArray> groupFieldsAndData = collectGroupFieldsAndData(data, groupFields, groupType, splitChar);

        for (Map.Entry<String, JSONArray> entry : groupFieldsAndData.entrySet()) {
            JSONArray groupData = entry.getValue();
            JSONObject resItem = groupData.getJSONObject(0);
            for (int i = 0; i < groupData.size(); i++) {
                JSONObject item = groupData.getJSONObject(i);
                for (int j = 0; j < statisticsFields.size(); j++) {
                    String statisticsField = statisticsFields.getString(j);
                    String newField1 = statisticsField + "_sum";
                    String newField2 = statisticsField + "_count";
                    String newField3 = statisticsField + "_avg";
                    resItem.put(newField1, resItem.getIntValue(newField1) + item.getIntValue(statisticsField));
                    resItem.put(newField2, resItem.getIntValue(newField2) + 1);
                    resItem.put(newField3, resItem.getIntValue(newField1) / (resItem.getIntValue(newField2) * 1.0));
                }
            }
            res.add(resItem);
        }

        return res;
    }


    public static Map<String, JSONArray> collectGroupFieldsAndData(JSONArray data, JSONArray groupFields, String groupType, String splitChar) {
        Map<String, JSONArray> groupFieldsAndData = new HashMap<>();
        Set<String> groupFieldsKeySet = groupFieldsAndData.keySet();


        if ("模糊匹配".equals(groupType)) {
            for (int i = 0; i < groupFields.size(); i++) {
                String field = groupFields.getString(i);
                for (int j = 0; j < data.size(); j++) {
                    JSONObject item = data.getJSONObject(j);
                    String value = item.getString(field);
                    if (value != null && value.contains(splitChar)) {
                        String[] valueSplit = value.split(splitChar);
                        // 如果需要去除未匹配数据，在此处加标识位
                        for (String s : valueSplit) {
                            JSONObject newItem = (JSONObject) item.clone();
                            newItem.put(field, s);
                            data.add(newItem);
                        }
                        data.remove(j);
                    }
                }
            }

        }

        for (int i = 0; i < data.size(); i++) {
            JSONObject item = data.getJSONObject(i);
            StringBuilder itemGroupFieldsValue = new StringBuilder();
            for (int j = 0; j < groupFields.size(); j++) {
                String field = groupFields.getString(j);
                itemGroupFieldsValue.append(item.get(field)).append(":");
            }
            groupFieldsAndData.putIfAbsent(itemGroupFieldsValue.toString(), new JSONArray());
            JSONArray groupData = groupFieldsAndData.get(itemGroupFieldsValue.toString());
            groupData.add(item);
        }

        return groupFieldsAndData;
    }

    @Override
    public SparkNodeEnum getType() {
        return SparkNodeEnum.MATCH;
    }
}
